package com.hmdp.service.impl;

import cn.hutool.core.bean.BeanUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.hmdp.dto.Result;
import com.hmdp.entity.SeckillVoucher;
import com.hmdp.entity.VoucherOrder;
import com.hmdp.mapper.VoucherOrderMapper;
import com.hmdp.service.IVoucherOrderService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hmdp.utils.RedisIDWorker;
import com.hmdp.utils.UserHolder;
import com.hmdp.utils.lock.impl.SimpleRedisLock;
import lombok.extern.slf4j.Slf4j;
import org.springframework.aop.framework.AopContext;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

/**
 * <p>
 *  服务实现类
 * </p>
 *
 * @author 虎哥
 * @since 2021-12-22
 */
@Service
@Slf4j
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {

    @Resource
    private SeckillVoucherServiceImpl seckillVoucherService;

    @Resource
    private RedisIDWorker redisIDWorker;

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    @Override
    public Result seckillVoucher(Long voucherId) {
        // 查优惠券
        SeckillVoucher voucher = seckillVoucherService.getById(voucherId);

        // 查时间是否开始
        if(voucher.getBeginTime().isAfter(LocalDateTime.now())){
            // 未开始
            return Result.fail("秒杀未开始");
        }
        if(voucher.getEndTime().isBefore(LocalDateTime.now())){
            return Result.fail("秒杀已结束");
        }
        // 判断库存是否充足
        if(voucher.getStock() < 1){
            return Result.fail("库存不足");
        }
        boolean success = seckillVoucherService.update().setSql("stock = stock - 1").eq("voucher_id", voucherId).update();
        if(!success){
            return Result.fail("更新数据库失败");
        }
        //创建订单
        VoucherOrder voucherOrder = new VoucherOrder();
        long orderId = redisIDWorker.nextId("order");
        voucherOrder.setId(orderId);
        voucherOrder.setUserId(UserHolder.getUser().getId());
        voucherOrder.setVoucherId(voucherId);
        this.save(voucherOrder);

        // 返回订单ID:
        return Result.ok(orderId);
    }

    // 乐观锁版本
    public Result seckillVoucher_v2(Long voucherId) {
        // 查优惠券
        SeckillVoucher voucher = seckillVoucherService.getById(voucherId);

        // 查时间是否开始
        if(voucher.getBeginTime().isAfter(LocalDateTime.now())){
            // 未开始
            return Result.fail("秒杀未开始");
        }
        if(voucher.getEndTime().isBefore(LocalDateTime.now())){
            return Result.fail("秒杀已结束");
        }
        // 判断库存是否充足
        int stockNum = voucher.getStock();
        if(stockNum< 1){
            return Result.fail("库存不足");
        }
        // 添加乐观锁 CAS法
        boolean success = seckillVoucherService.update().setSql("stock = stock - 1")
                .eq("stock", voucher.getStock()).eq("voucher_id", voucherId).update();
        if(!success){
            return Result.fail("更新数据库失败");
        }
        //创建订单
        VoucherOrder voucherOrder = new VoucherOrder();
        long orderId = redisIDWorker.nextId("order");
        voucherOrder.setId(orderId);
        voucherOrder.setUserId(UserHolder.getUser().getId());
        voucherOrder.setVoucherId(voucherId);
        this.save(voucherOrder);

        // 返回订单ID:
        return Result.ok(orderId);
    }


    // 实现一人一单 分布式方法
    public Result seckillVoucher_v3(Long voucherId) {
        // 查优惠券
        SeckillVoucher voucher = seckillVoucherService.getById(voucherId);

        // 查时间是否开始
        if(voucher.getBeginTime().isAfter(LocalDateTime.now())){
            // 未开始
            return Result.fail("秒杀未开始");
        }
        if(voucher.getEndTime().isBefore(LocalDateTime.now())){
            return Result.fail("秒杀已结束");
        }
        // 判断库存是否充足
        int stockNum = voucher.getStock();
        if(stockNum< 1){
            return Result.fail("库存不足");
        }

        Long userId = UserHolder.getUser().getId();
        synchronized (userId.toString().intern()){
            IVoucherOrderService proxy = (IVoucherOrderService)AopContext.currentProxy();
            return proxy.createVoucherOrder(voucherId);
        }
    }

    // 分布式场景下 实现一人一单 使用Redis分布式锁解决
    public Result seckillVoucher_v4(Long voucherId) {
        // 查优惠券
        SeckillVoucher voucher = seckillVoucherService.getById(voucherId);

        // 查时间是否开始
        if(voucher.getBeginTime().isAfter(LocalDateTime.now())){
            // 未开始
            return Result.fail("秒杀未开始");
        }
        if(voucher.getEndTime().isBefore(LocalDateTime.now())){
            return Result.fail("秒杀已结束");
        }
        // 判断库存是否充足
        int stockNum = voucher.getStock();
        if(stockNum< 1){
            return Result.fail("库存不足");
        }

        Long userId = UserHolder.getUser().getId();

        // 自己创建分布式锁对象
        SimpleRedisLock lock = new SimpleRedisLock(stringRedisTemplate,"order:" + userId);
        boolean isLocked = lock.tryLock(1200);

        if(!isLocked){
            // 获取锁失败
            return Result.fail("获取锁失败");
        }
        try{
            IVoucherOrderService proxy = (IVoucherOrderService)AopContext.currentProxy();
            return proxy.createVoucherOrder(voucherId);
        }finally {
            // 释放锁逻辑
            lock.unlock();
        }
    }

    /**
     * 线程池 用以实现阻塞队列优化异步秒杀
     */
    private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();


    // @PostConstruct注解使该方法在类初始化后就开始执行
    @PostConstruct
    private void init(){
        // SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());

        log.info("开启线程 处理消息队列");
        // redis消息队列版本
        SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler_stream());
    }


    /**
     * 线程任务: 不断从阻塞队列中获取订单
     */
    private class VoucherOrderHandler implements Runnable {


        @Override
        public void run() {

            while (true){
                try {
                    // 取任务
                    log.info("get task");
                    VoucherOrder voucherOrder = ordersTask.take();
                    // 执行任务
                    handleVoucherOrder(voucherOrder);
                } catch (InterruptedException e) {
                    log.error("异常", e);
                    throw new RuntimeException(e);
                }

            }
        }
    }

    private class VoucherOrderHandler_stream implements Runnable {
        @Override
        public void run() {
            log.info("处理消息队列");
            while (true){
                try {
                    // 获取消息队列中的订单信息
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                            Consumer.from("g1", "c1"),
                            StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                            StreamOffset.create("stream.orders", ReadOffset.lastConsumed())
                    );
                    // 2、判断消息获取是否成功
                    if (list == null || list.isEmpty()) {
                        // 2.1 消息获取失败，说明没有消息，进入下一次循环获取消息
                        continue;
                    }
                    // 获取成功可以下单
                    // 3、消息获取成功，可以下单
                    // 将消息转成VoucherOrder对象
                    MapRecord<String, Object, Object> record = list.get(0);
                    Map<Object, Object> messageMap = record.getValue();
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(messageMap, new VoucherOrder(), true);
                    handleVoucherOrder(voucherOrder);
                    // 4、ACK确认 SACK stream.orders g1 id
                    stringRedisTemplate.opsForStream().acknowledge("stream.orders", "g1", record.getId());
                    log.info("ack确认成功");
                } catch (Exception e) {
                    // 出异常了，ACK没确认 去pending-list处理
                    log.error("异常", e);
                    handlePendingList();
                    throw new RuntimeException(e);
                }

            }
        }
    }

    private void handlePendingList() {
        while (true) {
            try {
                // 1、从pendingList中获取订单信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 1000 STREAMS streams.order 0
                List<MapRecord<String, Object, Object>> messageList = stringRedisTemplate.opsForStream().read(
                        Consumer.from("g1", "c1"),
                        StreamReadOptions.empty().count(1),
                        StreamOffset.create("stream.orders", ReadOffset.from("0"))
                );
                // 2、判断pendingList中是否有效性
                if (messageList == null || messageList.isEmpty()) {
                    // 2.1 pendingList中没有消息，直接结束循环
                    break;
                }
                // 3、pendingList中有消息
                // 将消息转成VoucherOrder对象
                MapRecord<String, Object, Object> record = messageList.get(0);
                Map<Object, Object> messageMap = record.getValue();
                VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(messageMap, new VoucherOrder(), true);
                handleVoucherOrder(voucherOrder);
                // 4、ACK确认 SACK stream.orders g1 id
                stringRedisTemplate.opsForStream().acknowledge("stream.orders", "g1", record.getId());
            } catch (Exception e) {
                log.error("处理订单异常", e);
                // 这里不用调自己，直接就进入下一次循环，再从pendingList中取，这里只需要休眠一下，防止获取消息太频繁
                try {
                    Thread.sleep(20);
                } catch (InterruptedException ex) {
                    log.error("线程休眠异常", ex);
                }
            }
        }
    }


    private void handleVoucherOrder(VoucherOrder voucherOrder){
        //创建订单 不能再userHoder了 因为现在是多线程 ThradLocal下取不到主线程的东西
        // Long userId = UserHolder.getUser().getId();
        log.info("handle");
        Long userId = voucherOrder.getUserId();
        // 自己创建分布式锁对象
        SimpleRedisLock lock = new SimpleRedisLock(stringRedisTemplate,"order:" + userId);
        boolean isLocked = lock.tryLock(1200);

        if(!isLocked){
            log.info("获取锁失败");
        }
        try{
            log.info("threadProxy.createVoucherOrder_v2(voucherOrder);");
            threadProxy.createVoucherOrder_v2(voucherOrder);
        }finally {
            // 释放锁逻辑
            lock.unlock();
        }
    }

    private IVoucherOrderService threadProxy;

    private BlockingQueue<VoucherOrder> ordersTask = new ArrayBlockingQueue<>(1024 * 1024);

    /**
     * 加载 判断秒杀券库存是否充足 并且 判断用户是否已下单 的Lua脚本
     */
    private static final DefaultRedisScript<Long> SECKILL_SCRIPT;

    static {
        SECKILL_SCRIPT = new DefaultRedisScript<>();
        SECKILL_SCRIPT.setLocation(new ClassPathResource("lua/seckill.lua"));
        SECKILL_SCRIPT.setResultType(Long.class);
    }

    // 异步秒杀优化版本 基于lua和多线程优化异步秒杀效率 阻塞队列版本
    public Result seckillVoucher_v5(Long voucherId) {
        // 查优惠券等操作 放到lua脚本里面写
        Long userId = UserHolder.getUser().getId();

        // 1、执行Lua脚本，判断用户是否具有秒杀资格
        Long result = stringRedisTemplate.execute(
                    SECKILL_SCRIPT,
                    Collections.emptyList(),
                    voucherId.toString(), userId.toString());
        int r = result.intValue();
        if(r != 0){
            // 脚本执行结果不为0 无购买资格
            return Result.fail(r == 2 ? "不能重复下单" : "库存不足");
        }
        // 下单信息保存到阻塞队列里
        long orderId = redisIDWorker.nextId("order");
        VoucherOrder voucherOrder = new VoucherOrder();
        voucherOrder.setVoucherId(voucherId);
        voucherOrder.setUserId(userId);
        threadProxy = (IVoucherOrderService)AopContext.currentProxy();
        log.info("add task");
        ordersTask.add(voucherOrder);

        return Result.ok(orderId);


    }

    private static final DefaultRedisScript<Long> SECKILL_SCRIPT_v2;

    static {
        SECKILL_SCRIPT_v2 = new DefaultRedisScript<>();
        SECKILL_SCRIPT_v2.setLocation(new ClassPathResource("lua/stream-seckill.lua"));
        SECKILL_SCRIPT_v2.setResultType(Long.class);
    }

    // 异步秒杀优化版本 基于lua和多线程优化异步秒杀效率 消息队列版本
    public Result seckillVoucher_v6(Long voucherId) {

        // 查优惠券等操作 放到lua脚本里面写
        Long orderId = redisIDWorker.nextId("order");
        Long userId = UserHolder.getUser().getId();


        // 1、执行Lua脚本，判断用户是否具有秒杀资格
        log.info("执行lua脚本 添加到redis消息队列");
        Long result = stringRedisTemplate.execute(
                SECKILL_SCRIPT_v2,
                Collections.emptyList(),
                voucherId.toString(), userId.toString(), orderId.toString());
        int r = result.intValue();
        // 如果r = 0 说明有购买资格 待处理事件已放入消息队列
        if(r != 0){
            // 脚本执行结果不为0 无购买资格
            return Result.fail(r == 2 ? "不能重复下单" : "库存不足");
        }

        VoucherOrder voucherOrder = new VoucherOrder();
        voucherOrder.setVoucherId(voucherId);
        voucherOrder.setUserId(userId);
        threadProxy = (IVoucherOrderService)AopContext.currentProxy();

        return Result.ok(orderId);


    }

    @Transactional
    @Override
    public Result createVoucherOrder(Long voucherId) {

        Long userId = UserHolder.getUser().getId();
        // 1、判断当前用户是否是第一单
        int count = this.count(new LambdaQueryWrapper<VoucherOrder>()
                .eq(VoucherOrder::getUserId, userId));
        if (count >= 1) {
            // 当前用户不是第一单
            log.error("当前用户不是第一单");
            return Result.fail("当前用户不是第一单");
        }
        // 2、用户是第一单，可以下单，秒杀券库存数量减一
        boolean flag = seckillVoucherService.update(new LambdaUpdateWrapper<SeckillVoucher>()
                .eq(SeckillVoucher::getVoucherId, voucherId)
                .gt(SeckillVoucher::getStock, 0)
                .setSql("stock = stock -1"));
        if (!flag) {
            throw new RuntimeException("秒杀券扣减失败");
        }
        // 3、创建订单，并将订单保存到数据库
        VoucherOrder voucherOrder = new VoucherOrder();
        long orderId = redisIDWorker.nextId("order");
        voucherOrder.setId(orderId);
        voucherOrder.setUserId(userId);
        voucherOrder.setVoucherId(voucherId);
        flag = this.save(voucherOrder);
        if (!flag) {
            throw new RuntimeException("创建秒杀券订单失败");
        }
        return Result.ok(orderId);
    }

    // v2版本 适配阻塞队列的异步秒杀下单优化
    @Transactional
    public void createVoucherOrder_v2(VoucherOrder voucherOrder) {
        log.info("come in threadProxy.createVoucherOrder_v2(voucherOrder);");
        Long userId = voucherOrder.getUserId();
        // 1、判断当前用户是否是第一单
        int count = this.count(new LambdaQueryWrapper<VoucherOrder>()
                .eq(VoucherOrder::getUserId, userId));
        if (count >= 1) {
            // 当前用户不是第一单
            log.error("当前用户不是第一单");
        }
        log.info("voucherId=" + voucherOrder.getVoucherId());
        // 将订单保存到数据库
        try {
            boolean flag = seckillVoucherService.update(new LambdaUpdateWrapper<SeckillVoucher>()
                    .eq(SeckillVoucher::getVoucherId, voucherOrder.getVoucherId())
                    .gt(SeckillVoucher::getStock, 0)
                    .setSql("stock = stock -1"));
            if (!flag) {
                throw new RuntimeException("秒杀券扣减失败");
            }
            long orderId = redisIDWorker.nextId("order");
            voucherOrder.setId(orderId);
            flag = this.save(voucherOrder);
            log.info("保存订单成功");
            if (!flag) {
                throw new RuntimeException("创建秒杀券订单失败");
            }
        } catch (Exception e) {
            throw e;
        }
    }


}
